package br.biofoco.p2p.rpc.tcp;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import br.biofoco.p2p.rpc.EndPoint;
import br.biofoco.p2p.rpc.Protocol;
import br.biofoco.p2p.rpc.messaging.CommandHandler;
import br.biofoco.p2p.rpc.messaging.WireRequestMessage;
import br.biofoco.p2p.rpc.messaging.Transport;
import br.biofoco.p2p.rpc.messaging.TransportClient;
import br.biofoco.p2p.utils.Serializer;
import br.biofoco.p2p.utils.SerializerFactory;

public class TcpTransportServer implements Transport {

	private ServerThread serverThread;
	private volatile boolean active;
	public static final int DEFAULT_PORT = 7171;
	private final ExecutorService executor = Executors.newFixedThreadPool(10);
	private CommandHandler commandHandler;
	
	private Serializer serializer = SerializerFactory.getSerializer();
	private int port;

	private static final Logger LOGGER = LoggerFactory.getLogger(TcpTransportServer.class);
	
	public TcpTransportServer() {
		this.port = DEFAULT_PORT;
	}
	
	public TcpTransportServer(int port) {
		this.port = port;
	}
	
	@Override
	public void start() throws IOException {
		if (!active) {
			LOGGER.debug("Starting TCP transport server");

			active = true;
			serverThread = new ServerThread();
			serverThread.startServer();
		}
	}

	@Override
	public void stop() throws IOException {
		if (active) {
			LOGGER.debug("Stopping TCP transport server");
			serverThread.stopServer();
		}
	}

	public boolean isActive() {
		return active;
	}

	@Override
	public TransportClient createClient(EndPoint endpoint) throws IOException {
		TcpEndPoint tcpEndpoint = (TcpEndPoint) endpoint;
		Socket socket = new Socket(tcpEndpoint.getAddress(), tcpEndpoint.getPort());
		return new TcpTransportClient(socket);
	}

	private class ServerThread implements Runnable {

		private ServerSocket server;
		//TODO BUG ridden code!!! The list has to be cleaned from time to time and it must be clean once the client goes out.
		private List<ClientThread> clients = new ArrayList<ClientThread>();

		public void startServer() {
			executor.execute(this);
		}

		public void stopServer() {
			active = false;
			closeAllConnections();
		}

		@Override
		public void run() {
			try {
				active = true;
				server = new ServerSocket(port);
				LOGGER.debug("TCP transport listening on port " + port);

				while (active) {
					try {
						ClientThread ct = new ClientThread(server.accept());
						executor.execute(ct);
						clients.add(ct);
						LOGGER.debug("TCP transport client connected");
					} catch (IOException ie) {
						LOGGER.error(ie.getMessage());
					}
				}

			} catch (IOException e) {
				active = false;
				//e.printStackTrace();
			}

			closeAllConnections();
		}

		private void closeAllConnections() {
			try {
				for (ClientThread pendingClient : clients) {
					if (pendingClient.isRunning())
						pendingClient.closeClientConnection();
				}
				if (server != null)
					server.close();
			} catch (IOException ie) {
				LOGGER.info(ie.getMessage());
			} finally {
				executor.shutdownNow();
			}
		}

	}

	private class ClientThread implements Runnable {

		private final Socket socket;
		private volatile boolean running;

		public ClientThread(Socket socket) throws IOException {
			this.socket = socket;
			running = true;
		}

		public boolean isRunning() {
			return running;
		}

		public void run() {
			try {				
				while (running) {						
					WireRequestMessage request = serializer.unserialize(socket.getInputStream(), WireRequestMessage.class);				

					if (request.getMethod().equals("CLOSE")){
						running = false;
						break;
					}					
					
					if (commandHandler == null) {
						LOGGER.warn("There's no provided Protocol Handler, resorting to DEFAULT protocol");
						commandHandler = new DefaultProtocolHandler();
						
					}
					
					serializer.serialize(commandHandler.doProcess(request),socket.getOutputStream());
				}
			} catch (IOException ie) {
				// client closed connection abruptly!
				//ie.printStackTrace();
				LOGGER.error(ie.getMessage());
				running = false;
			} 
			finally {
				closeClientConnection();
			}
		}

		public void closeClientConnection() {
			try {
				LOGGER.info("Closing client connection");				
				running = false;
				socket.close();
			} catch (IOException ioe) {
				LOGGER.error(ioe.getMessage());
				//ioe.printStackTrace();
			}
		}
	}

	@Override
	public void setCommandHandler(CommandHandler handler) {		
		commandHandler = handler;		
	}

	@Override
	public Protocol getProtocol() {
		return Protocol.TCP;
	}
}
